package raft

//实现一个简易的raft算法程序
import (
	"encoding/gob"
	"fmt"
	"log"
	"math/rand"
	"net/rpc"
	"os"
	"path/filepath"
	"sync"
	"time"
)

const (
	nodenum = 3
)
const (
	follower  = 0
	candidate = 1
	leader    = 2
)
const (
	node0port = 50051
	node1port = 50052
	node2port = 50053
)

var (
	DEBUG                     = true
	LOG_HEARTBEAT             = false
	LOG_TIMEOUT               = true
	LOG_LOCK                  = false
	DISABLE_PERSIST           = true
	DISABLE_NEXTINDEX_BIGSTEP = true // disable the optimization for big step to decrease nextIndex
	ENABLE_DISCONNECT         = true
)

// Raft’s RPCs typically require the recipient to persist information to stable storage,
// so the broadcast time may range from 0.5ms to 20ms, depending on storage technology.
// As a result, the election timeout is likely to be somewhere between 10ms and 500ms.

const (
	unvoted              = -1
	notfoundConflictTerm = -1
	heartbeatTimeout     = time.Duration(1000) * time.Millisecond
	applyTimeout         = time.Duration(1000) * time.Millisecond
	electionTimeoutMin   = 4000 * time.Millisecond
	electionTimeoutMax   = 6000 * time.Millisecond
	LOG_CLASS_HEARTBEAT  = "RPC:Heart"
	LOG_CLASS_APPEND     = "RPC:Entry"
	LOG_CLASS_VOTE       = "RPC:Voted"
	LOG_CLASS_CLIENT     = "RPC:Clien"
	LOG_CLASS_LOCK       = "Lock"
	LOG_CLASS_TIMEOUT    = "Time"
	LOG_CLASS_LIFECYCLE  = "Life"
	LOG_CLASS_COMMIT     = "Cmit"
	LOG_CLASS_LEADING    = "Lead"
	LOG_CLASS_ELECTION   = "Elec"
)

// 四个节点的ipv4
// 需要用到的工具函数
// 生成随机的等待超时时间
func minInt(a int, b int) int {
	if a < b {
		return a
	} else {
		return b
	}
}
func getRandomizedElectionTimeout() time.Duration {
	return electionTimeoutMin + time.Duration(rand.Int63())%(electionTimeoutMax-electionTimeoutMin)
}

// LogValue represents a database operation request
type LogValue struct {
	Data      interface{}            `json:"data"`       // The actual data for insert/update operations
	Operation string                 `json:"operation"`  // Operation type: "insert", "update", "delete", "select"
	Condition map[string]interface{} `json:"condition"`  // SQL WHERE clause for update/delete operations
	TableName string                 `json:"table_name"` // Target table name
}

// 日志
type LogEntry struct {
	Term  int
	Value LogValue
}
type Raft struct {
	Mu sync.Mutex //分布式锁实现
	//Persister   *Persister    //用来保存状态，存储
	Me          int         //当前节点在其他节点中的索引
	Pieces      int         //分片号
	Logger      *log.Logger //记录日志
	Hearted     int         //心跳计数
	CurrentTerm int         //当前节点已知的最新任期号
	VotedFor    int         //在当前任期内投票给了哪个candidater(-1表示没有投票)
	Logs        []LogEntry  //日志数组，索引从1开始
	CommitIndex int         //已知已经提交的最高日志条目索引
	LastApplied int         //已经应用到状态机的最高日志条目索引
	//leader专用数据，选举后重新初始化
	NextIndex      []int         //每个副本节点的下一个要发送的日志索引
	MatchIndex     []int         //每个副本节点已经复制的最高日志索引
	Connected      []bool        //记录每个副本节点的连接状态
	VoteGranted    []bool        //记录在选举中获得的投票情况
	State          int           //节点当前的状态，follower=0，candidate=1,leader=2
	HeartbeatTimer *time.Timer   //心跳定时器
	ElectionTimer  *time.Timer   //选举超时定时器
	ApplyTimer     *time.Timer   //应用日志到状态机的定时器
	ApplyCh        chan ApplyMsg //应用消息通道
	KillCh         chan struct{}
	Locked         bool //节点是否被锁定
}

// candidate向其他节点发送的数据结构体
type RequestVoteArgs struct {
	PieceNum     int
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

// 其他节点给candidate返回的结构体
type RequestVoteReply struct {
	PieceNum    int
	Term        int
	VoteGranter bool
}
type ApplyMsg struct {
	Index   int
	Command LogValue
}

// Make creates a new Raft instance
func Make(nodeNum int, me int, piece int, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.Me = me
	rf.Pieces = piece
	rf.ApplyCh = applyCh
	rf.Hearted = 0
	rf.CurrentTerm = 0
	rf.VotedFor = -1 // unvoted

	rf.Logs = make([]LogEntry, 0)
	rf.Logs = append(rf.Logs, LogEntry{}) // index 0 is unused

	rf.CommitIndex = 0
	rf.LastApplied = 0

	rf.NextIndex = make([]int, nodeNum)
	rf.MatchIndex = make([]int, nodeNum)
	rf.Connected = make([]bool, nodeNum)
	rf.VoteGranted = make([]bool, nodeNum)

	rf.HeartbeatTimer = time.NewTimer(heartbeatTimeout)
	rf.ElectionTimer = time.NewTimer(getRandomizedElectionTimeout())
	rf.ApplyTimer = time.NewTimer(time.Second)

	rf.State = 0 // follower

	// 初始化日志记录器
	rf.Logger = log.New(os.Stdout, fmt.Sprintf("[Raft%d-%d] ", me, piece), log.Lmicroseconds)
	rf.KillCh = make(chan struct{})

	// initialize from state persisted before a crash
	//	rf.readPersist(persister.ReadRaftState())
	fmt.Printf("节点 %d 分片 %d 的 Raft 实例创建成功\n", me, piece)

	// 设置调试开关为 true 以显示日志
	DEBUG = true
	// 计算数据库绝对路径，避免在不同工作目录下创建新的空库
	cwd, _ := os.Getwd()
	dbfile := fmt.Sprintf("db%d.sqlite3", me)
	candidates := []string{
		filepath.Join(cwd, dbfile),
		filepath.Join(cwd, "..", dbfile),
		filepath.Join(cwd, "..", "..", dbfile),
	}
	dbname := ""
	for _, p := range candidates {
		if _, err := os.Stat(p); err == nil {
			dbname = p
			break
		}
	}
	if dbname == "" {
		// 回退到项目根目录的推断路径
		projectRoot := filepath.Clean(filepath.Join(cwd, "..", ".."))
		dbname = filepath.Join(projectRoot, dbfile)
	}
	applyerInstance, err := NewApplyer(dbname, me, 0)
	if err != nil {
		log.Fatalf("初始化 Applyer 失败: %v", err)
	}
	// 在单独的 goroutine 中消费 ApplyCh，避免阻塞 Make()
	go func() {
		defer applyerInstance.Close()
		for {
			select {
			case msg, ok := <-rf.ApplyCh:
				if !ok {
					return
				}
				if err := applyerInstance.Apply(msg); err != nil {
					log.Printf("应用日志失败: %v", err)
				}
			case <-rf.KillCh:
				return
			}
		}
	}()
	go rf.longrun()

	return rf
}

// 确保 RPC 可以序列化/反序列化日志中的 interface{} 值
// AppendEntriesArgs.Entries 里的每个 LogEntry.Value 是 interface{}
// 当该值实际是 LogValue 时，gob 需要提前注册具体类型。
func init() {
	gob.Register(LogValue{})
	// 常见的通用载体，也注册一下，避免 Data 等字段为 interface{} 时失败
	gob.Register(map[string]interface{}{})
	gob.Register([]interface{}{})
}
func (rf *Raft) longrun() {
	rf.LogClass(LOG_CLASS_LIFECYCLE, "主循环开始运行...")
	fmt.Printf("[Raft%d-%d] 主循环开始运行...\n", rf.Me, rf.Pieces)
	go func() {
		for {
			select {
			case <-rf.ElectionTimer.C: //如果选举计时器到点了
				go func() {
					if rf.hasKilled() {
						return
					}
					rf.InLock(func() { rf.campaign() }, "Long-run campaign")
				}()
			case <-rf.HeartbeatTimer.C: //如果心跳到时间了
				go func() {
					if rf.hasKilled() {
						return
					}
					rf.InLock(func() { rf.heartbeat() }, "Long-run beartbeat")
				}()
			case <-rf.KillCh:
				return
			}
			time.Sleep(100 * time.Millisecond)
		}
	}()
	go func() {
		for {
			select {
			case <-rf.ApplyTimer.C:
				go func() {
					if rf.hasKilled() {
						return
					}
					rf.InLock(func() { rf.apply() }, "Long-run applying")
				}()
				rf.ApplyTimer.Reset(applyTimeout)
			case <-rf.KillCh:
				return

			}
			//time.Sleep(1000 * time.Millisecond)
		}
	}()
}

// 日志记录函数
func (rf *Raft) LogClass(class string, format string, v ...any) {
	if !DEBUG {
		return
	}

	if !LOG_HEARTBEAT && class == LOG_CLASS_HEARTBEAT {
		return
	}

	if !LOG_TIMEOUT && class == LOG_CLASS_TIMEOUT {
		return
	}

	if !LOG_LOCK && class == LOG_CLASS_LOCK {
		return
	}

	rf.Log("(%s) %s", class, fmt.Sprintf(format, v...))
}

// 节点状态日志打印函数
func (rf *Raft) Log(format string, v ...interface{}) {
	if !DEBUG {
		return
	}
	role := "follow"
	switch rf.State {
	case candidate:
		role = "candid"
	case leader:
		role = "leader"
	}
	locked := "-"
	// if rf.Locked {
	//     locked = "+"
	// }
	rf.Logger.Printf("%d%s(%s)[%d,%d>%d>%d]: %s\n", rf.Me, locked, role, rf.CurrentTerm, rf.LastApplied, rf.CommitIndex, rf.len(), fmt.Sprintf(format, v...))
}
func (rf *Raft) hasKilled() bool { //监听某个节点有没有寄掉
	select {
	case <-rf.KillCh:
		return true
	default:
		return false
	}
}

// 如果选举超时了，开始竞争！
func (rf *Raft) campaign() {
	if rf.State == leader || rf.hasKilled() {
		return
	}
	rf.LogClass(LOG_CLASS_LIFECYCLE, "节点%d中的分片%d 竞争任期 %d", rf.Me, rf.Pieces, rf.CurrentTerm+1)
	for i := range rf.VoteGranted {
		rf.VoteGranted[i] = false
	}
	rf.State = candidate
	rf.CurrentTerm++    //当选择竞选时，给自己任期加1
	rf.VotedFor = rf.Me //每个candidate先给自己投票
	rf.resetElectionTimeout()
	currentTerm := rf.CurrentTerm
	index, term := rf.lastLogSignature()
	args := RequestVoteArgs{
		PieceNum:     rf.Pieces,
		Term:         currentTerm,
		CandidateId:  rf.Me,
		LastLogIndex: index,
		LastLogTerm:  term,
	}
	//通过rpc向其他节点发送选举消息
	for i := 0; i < nodenum; i++ {
		if i == rf.Me {
			continue
		}
		go func(i int) {
			reply := RequestVoteReply{}
			ok := false
			for !ok && rf.State == candidate && rf.CurrentTerm == currentTerm {
				if rf.hasKilled() {
					return
				}
				ok = rf.sendRequestVote(i, args, &reply) //这个函数里边封装不同节点间发送rpc
			}
			if !(rf.State == candidate && rf.CurrentTerm == currentTerm) {
				return
			}
			rf.checkFollow(args)
			if rf.State != candidate {
				return
			}
			if reply.VoteGranter {
				rf.LogClass(LOG_CLASS_ELECTION, "%d 收到来自%d 的投票 在任期%d", rf.Me, i, currentTerm)
				rf.VoteGranted[i] = true
				if rf.isWinner() {
					rf.LogClass(LOG_CLASS_ELECTION, "%d 在任期%d 中赢得了选举", rf.Me, currentTerm)
					rf.lead()
				}
			} else {
				rf.LogClass(LOG_CLASS_ELECTION, "节点%d 在任期 %d的选举中没有获得来自%d 的投票", rf.Me, currentTerm, i)
			}
		}(i)
	}
}
func (rf *Raft) resetElectionTimeout() { //重置选举时间
	timeout := getRandomizedElectionTimeout()
	rf.LogClass(LOG_CLASS_TIMEOUT, "%d 重置当前节点的选举时间：%d", rf.Me, timeout.Milliseconds())
	// 安全地停止并清空计时器通道后重置
	if !rf.ElectionTimer.Stop() {
		select {
		case <-rf.ElectionTimer.C:
		default:
		}
	}
	rf.ElectionTimer.Reset(timeout)

}

// 返回上一次提交的日志索引和任期
func (rf *Raft) lastLogSignature() (int, int) {
	index := rf.len()
	term := 0
	if index >= 1 {
		term = rf.Logs[index].Term
	}
	return index, term
}

// 返回日志长度
func (rf *Raft) len() int {
	return len(rf.Logs) - 1
}

// 如果candidate接受到了更大的term则更新自己的term然后变成follow（没那个实力就给我趴着）
func (rf *Raft) checkFollow(data RequestVoteArgs) {
	if data.Term > rf.CurrentTerm {
		rf.CurrentTerm = data.Term
		rf.VotedFor = unvoted
		rf.follow()
	}
}
func (rf *Raft) checkFollowA(data AppendEntriesArgs) {
	if data.Term > rf.CurrentTerm {
		rf.CurrentTerm = data.Term
		rf.VotedFor = unvoted
		rf.follow()
	}
}
func (rf *Raft) follow() {
	if rf.State == follower {
		return
	}
	rf.LogClass(LOG_CLASS_LIFECYCLE, "%d follow at term %d", rf.Me, rf.CurrentTerm)
	rf.State = follower
	rf.VotedFor = unvoted
	rf.resetElectionTimeout()
}
func (rf *Raft) isWinner() bool {
	votedCount := 0
	for i, v := range rf.VoteGranted {
		if v || i == rf.Me {
			votedCount++
		}
	}
	rf.LogClass(LOG_CLASS_ELECTION, "%d 当前获得的票数是 %d", rf.Me, votedCount)
	return votedCount*2 > nodenum
}
func (rf *Raft) lead() {
	if rf.State == leader {
		return
	}
	rf.LogClass(LOG_CLASS_LIFECYCLE, "%d 在任期%d 中开始领导", rf.Me, rf.CurrentTerm)
	rf.State = leader
	lastIndex := rf.len()
	for i := 0; i < nodenum; i++ {
		rf.NextIndex[i] = lastIndex + 1
		rf.MatchIndex[i] = 0
		if ENABLE_DISCONNECT {
			rf.Connected[i] = true
		}
	}
	rf.heartbeat() //开始向其他节点发送心跳包
}
func (rf *Raft) heartbeat() {
	if rf.State != leader || rf.hasKilled() {
		return
	}
	rf.LogClass(LOG_CLASS_LIFECYCLE, "%d 在任期 %d 中开始发送心跳包", rf.Me, rf.CurrentTerm)
	lastLogIndex := rf.len()
	currentTerm := rf.CurrentTerm
	rf.Hearted++
	hearted := rf.Hearted
	for i := 0; i < nodenum; i++ {
		if i == rf.Me {
			continue
		}
		go func(i int) { //这里传入的是节点号
			reply := AppendEntriesReply{}
			retry := true
			for retry && (rf.State == leader) && (rf.CurrentTerm == currentTerm) && rf.Hearted == hearted {
				if rf.hasKilled() {
					return
				}
				var index, term int
				var args AppendEntriesArgs
				send := true
				//这里有一个rf.inlock
				if !(rf.State == leader && rf.CurrentTerm == currentTerm && rf.Hearted == hearted) {
					send = false
					return
				}
				index, term = rf.prevLogSignature(i)
				args = AppendEntriesArgs{
					Term:         currentTerm,
					LeaderId:     rf.Me,
					PrevLogIndex: index,
					PrevLogTerm:  term,
					Entries:      rf.Logs[index+1:],
					LeaderCommit: rf.CommitIndex,
				}
				//到这里结束
				ok := false
				if send {
					ok = rf.sendAppendEntries(i, args, &reply)
				} else {
					break
				}
				retry = !ok
				if !(rf.State == leader && rf.CurrentTerm == currentTerm && rf.Hearted == hearted) {
					return
				}
				if ENABLE_DISCONNECT {
					rf.Connected[i] = ok
					//如果丢失了大部分的连接，变成follower
					if !rf.isConnected() {
						rf.LogClass(LOG_CLASS_LIFECYCLE, "%d 在任期 %d 失去了连接变成了follower", rf.Me, currentTerm)
						rf.follow()
					}
				}
				if !ok {
					return
				}
				temr := AppendEntriesArgs{}
				temr.Term = args.Term
				rf.checkFollowA(temr)
				if rf.State != leader {
					return
				}
				if reply.Success {
					//如果成功，则对follower更新nextindex和matchindex
					if args.isheartbeat() {
						rf.LogClass(LOG_CLASS_LEADING, "成功向%d 发送了心跳包，nextindex是 %d,matchindex是 %d:%+v", i, rf.NextIndex[i], rf.MatchIndex[i], args)
					} else {
						if rf.MatchIndex[i] < lastLogIndex {
							rf.NextIndex[i] = lastLogIndex + 1
							rf.MatchIndex[i] = lastLogIndex
						}
						rf.LogClass(LOG_CLASS_LEADING, "AppendEntries success for %d, nextIndex -> %d, matchIndex -> %d: %+v", i, rf.NextIndex[i], rf.MatchIndex[i], args)

					}

				} else {
					if args.isheartbeat() {
						rf.LogClass(LOG_CLASS_LEADING, "Hearbeat fails at %d with term %d: response %+v, reply %+v", args.PrevLogIndex, args.PrevLogTerm, args, reply)
					} else {
						rf.LogClass(LOG_CLASS_LEADING, "AppendEntries fails at %d with term %d: response %+v, reply %+v", args.PrevLogIndex, args.PrevLogTerm, args, reply)
					}

					newNextIndex := index

					if newNextIndex < rf.NextIndex[i] {
						rf.NextIndex[i] = newNextIndex
						rf.LogClass(LOG_CLASS_LEADING, "nextIndex[%d]->%d", i, rf.NextIndex[i])
						retry = true
					} else {
						rf.LogClass(LOG_CLASS_LEADING, "nextIndex[%d]->%d(by another coroutine)", i, rf.NextIndex[i])
					}
				}
			}
		}(i)
	}
	if rf.State == leader {
		rf.resetHeartbeatTimeout()
	}
}

// 获取一个follower之前提交的日志的索引和任期
func (rf *Raft) prevLogSignature(server int) (int, int) {
	index := rf.NextIndex[server] - 1
	term := 0
	if index >= 1 {
		term = rf.Logs[index].Term
	}
	return index, term
}

// 重置心跳包时间
func (rf *Raft) resetHeartbeatTimeout() {
	rf.LogClass(LOG_CLASS_TIMEOUT, "%d 重置了心跳时间:%d", rf.Me, heartbeatTimeout.Milliseconds())
	// 安全地停止并清空计时器通道后重置
	if !rf.HeartbeatTimer.Stop() {
		select {
		case <-rf.HeartbeatTimer.C:
		default:
		}
	}
	rf.HeartbeatTimer.Reset(heartbeatTimeout)
}

// 将已经提交的日志应用到状态机（重要）
// 目前还是用通道机制，可以改为rpc调用
func (rf *Raft) apply() {
	if rf.State == leader {
		rf.updateCommitIndex()
	}
	for rf.CommitIndex > rf.LastApplied {
		cur := rf.LastApplied + 1
		msg := ApplyMsg{
			Index:   cur,
			Command: rf.Logs[cur].Value,
		}
		rf.LogClass(LOG_CLASS_COMMIT, "应用 %d:%+v", cur, msg)
		rf.ApplyCh <- msg
		rf.LastApplied = cur
		rf.LogClass(LOG_CLASS_COMMIT, "应用 %d:%+v", cur, msg)
	}
}

// 找最新可以提交的日志（如果被复制到大多数节点上则可以提交）
// 找当前任期的第一个日志然后判断
func (rf *Raft) updateCommitIndex() {
	newIndex := rf.CommitIndex + 1
	for newIndex <= rf.len() && rf.Logs[newIndex].Term != rf.CurrentTerm {
		newIndex++
	} //先找到是自己任期提交的日志
	if newIndex <= rf.len() {
		matchedCount := 0
		_ = matchedCount
	}
	if newIndex <= rf.len() {
		matchedCount := 0
		for _, v := range rf.MatchIndex {
			if v >= newIndex {
				matchedCount++
			}
		}
		if matchedCount*2 > nodenum {
			rf.CommitIndex = newIndex
			rf.LogClass(LOG_CLASS_COMMIT, "领导者的提交日志索引更新到了:%d", rf.CommitIndex)
		}
	}
}

// 获取当前节点的任期，是否为leader(是则返回true)
func (rf *Raft) GetState() (int, bool) {
	rf.Lock("GetState")
	defer rf.Unlock("GetState")
	return rf.CurrentTerm, rf.State == leader
}

// 锁住rf的mu，防止竞争
func (rf *Raft) Lock(name string) {
	rf.Mu.Lock()
	rf.Locked = true
	rf.LogClass(LOG_CLASS_LOCK, "将 %s 锁住", name)
}

// 解锁rf的mu
func (rf *Raft) Unlock(name string) {
	rf.Locked = false
	rf.Mu.Unlock()
	rf.LogClass(LOG_CLASS_LOCK, "将 %s 解锁", name)
}

// 锁封装函数
func (rf *Raft) InLock(f func(), name string) {
	rf.Lock(name)
	defer rf.Unlock(name)

	f()
}

// 解锁封装函数
func (rf *Raft) OutLock(f func(), name string) {
	rf.Unlock(name)
	defer rf.Lock(name)
	f()
}

type RpcTransferData interface {
	term() int
}
type AppendEntriesArgs struct {
	PieceNum     int
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []LogEntry
	LeaderCommit int
}

func (data AppendEntriesArgs) term() int {
	return data.Term
}

// 区分收到的这个包是心跳还是日志复制
func (data AppendEntriesArgs) isheartbeat() bool {
	return len(data.Entries) == 0
}

type AppendEntriesReply struct {
	Term          int
	Success       bool
	ConflictTerm  int
	ConflictIndex int
}

func (data AppendEntriesReply) term() int {
	return data.Term
}

// follower处理leader发来的日志复制请求的函数
// 这两个函数还需要再细改
func (rf *Raft) AppendEntries(args AppendEntriesArgs, reply *AppendEntriesReply) {
	if args.isheartbeat() {
		rf.LogClass(LOG_CLASS_HEARTBEAT, "收到来自%d的心跳包:%+v", args.LeaderId, args)
	} else {
		rf.LogClass(LOG_CLASS_APPEND, "收到来自 %d 的日志复制包:来自（%d,%d)带着 %d 日志： %+v", args.LeaderId, args.PrevLogIndex, args.PrevLogTerm, len(args.Entries), args)
	}
	rf.InLock(func() {
		rf.checkFollowA(args)
		//人家leader都发话了你一个candidate还狂啥
		if rf.State == candidate && rf.CurrentTerm == args.Term {
			rf.follow()
		}
		reply.Term = rf.CurrentTerm
		if args.Term < rf.CurrentTerm {
			reply.Success = false
		} else {
			rf.resetElectionTimeout()
			if args.PrevLogIndex > rf.len() {
				reply.Success = false
				if !DISABLE_NEXTINDEX_BIGSTEP {
					reply.ConflictIndex = rf.len() + 1
					reply.ConflictTerm = notfoundConflictTerm
				}
			} else {
				if args.PrevLogIndex >= 1 && rf.Logs[args.PrevLogIndex].Term != args.PrevLogTerm {
					reply.Success = false
					if !DISABLE_NEXTINDEX_BIGSTEP {
						reply.ConflictTerm = rf.Logs[args.PrevLogIndex].Term
						firstIndex := args.PrevLogIndex
						for firstIndex-1 >= 1 && rf.Logs[firstIndex-1].Term == reply.ConflictTerm {
							firstIndex--
						}
						reply.ConflictIndex = firstIndex
					}
					rf.Logs = rf.Logs[:args.PrevLogIndex]
				} else {
					reply.Success = true
					firstUnmatch := 0
					for firstUnmatch < len(args.Entries) && args.PrevLogIndex+1+firstUnmatch <= rf.len() && rf.Logs[args.PrevLogIndex+1+firstUnmatch].Term == args.Entries[firstUnmatch].Term {
						firstUnmatch++
					}
					if firstUnmatch <= len(args.Entries) {
						args.Entries = args.Entries[firstUnmatch:]
					}
					lastNewIndex := args.PrevLogIndex + firstUnmatch
					if len(args.Entries) > 0 {
						if lastNewIndex+1 <= rf.len() {
							rf.Logs = rf.Logs[:lastNewIndex+1]
						}
						rf.Logs = append(rf.Logs, args.Entries...)
						lastNewIndex += len(args.Entries)
					}
					if args.LeaderCommit > rf.CommitIndex {
						newCommitIndex := minInt(args.LeaderCommit, lastNewIndex)
						if newCommitIndex > rf.CommitIndex {
							rf.CommitIndex = newCommitIndex
							rf.LogClass(LOG_CLASS_COMMIT, "follower的提交日志更新到: %d", rf.CommitIndex)
						}
					}
				}
			}
		}
	}, LOG_CLASS_APPEND)
	//rf.persist()   //目前版本不需要永久化存储
	if args.isheartbeat() {
		rf.LogClass(LOG_CLASS_HEARTBEAT, "回复 %t 给 %d:%+v", reply.Success, args.LeaderId, args)
	} else {
		rf.LogClass(LOG_CLASS_APPEND, "回复 %t 给%d:来自(%d,%d) 带着 %d 的日志：%+v", reply.Success, args.LeaderId, args.PrevLogIndex, args.PrevLogTerm, len(args.Entries), args)
	}
}

// 通过rpc发送日志
func (rf *Raft) sendAppendEntries(server int, args AppendEntriesArgs, reply *AppendEntriesReply) bool {
	// Ensure the args carry the shard/piece number so the remote node can
	// dispatch to the correct Raft instance.
	// rf.Pieces holds this instance's piece id.
	args.PieceNum = rf.Pieces

	// Node RPC servers listen on ports 8080,8081,8082 (one per node).
	var port int
	switch server {
	case 0:
		port = node0port
	case 1:
		port = node1port
	case 2:
		port = node2port
	}
	address := fmt.Sprintf("localhost:%d", port)
	client, err := rpc.Dial("tcp", address)
	if err != nil {
		rf.LogClass(LOG_CLASS_APPEND, "发送给节点%d 日志失败，连接错误:%v", server, err)
		return false
	}
	defer client.Close()
	//这里通过RPC调用appendentries
	if len(args.Entries) == 0 {
		rf.LogClass(LOG_CLASS_HEARTBEAT, "发送给 %d: %+v", server, args)
	} else {
		rf.LogClass(LOG_CLASS_APPEND, "发送来自 (%d, %d) 带着 %d 日志给 %d: %+v", args.PrevLogIndex, args.PrevLogTerm, len(args.Entries), server, args)
	}
	// Call the node-level RPC handler which will forward to the correct
	// per-shard Raft instance based on args.PieceNum.
	callErr := client.Call("node.AppendEntries", args, reply)
	if callErr != nil {
		if len(args.Entries) == 0 {
			rf.LogClass(LOG_CLASS_HEARTBEAT, "Failed to send to %d: %+v", server, args)
		} else {
			rf.LogClass(LOG_CLASS_APPEND, "Failed to send from (%d, %d) with %d entries to %d: %+v", args.PrevLogIndex, args.PrevLogTerm, len(args.Entries), server, args)
		}
		return false
	}
	return true
}
func (data RequestVoteReply) term() int {
	return data.Term
}
func (rf *Raft) isUpToDate(lastLogIndex int, lastLogTerm int) bool {
	index, term := rf.lastLogSignature()
	return lastLogTerm > term || lastLogTerm == term && lastLogIndex >= index
}
func (rf *Raft) sendRequestVote(server int, args RequestVoteArgs, reply *RequestVoteReply) bool {
	rf.LogClass(LOG_CLASS_VOTE, "节点%d中的分片%d发送给%d请求选举包:%+v", rf.Me, rf.Pieces, server, args)
	var port int
	switch server {
	case 0:
		port = node0port
	case 1:
		port = node1port
	case 2:
		port = node2port
	}
	address := fmt.Sprintf("localhost:%d", port)
	client, err := rpc.Dial("tcp", address)
	if err != nil {
		rf.LogClass(LOG_CLASS_VOTE, "节点%d中的分片%d请求节点%d 投票失败，连接错误:%v", rf.Me, rf.Pieces, server, err)
		return false
	}
	defer client.Close()
	ok := client.Call("node.RequestVote", args, reply) == nil
	//这里通过RPC调用requestvote
	if !ok {
		rf.LogClass(LOG_CLASS_VOTE, "节点%d中的分片%d请求节点%d 投票失败:%+v", rf.Me, rf.Pieces, server, args)
	}
	return ok
}

// candidate向其他节点调用这个函数
func (rf *Raft) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) {
	rf.LogClass(LOG_CLASS_VOTE, "收到来自%d的数据包:%+v", args.CandidateId, args)
	rf.LogClass(LOG_CLASS_VOTE, "当前任期:%d,候选节点任期：%d,votedfor:%d", rf.CurrentTerm, args.Term, rf.VotedFor)
	lastIndex, lastTerm := 0, 0
	rf.checkFollow(args)
	reply.Term = rf.CurrentTerm
	reply.VoteGranter = false
	lastIndex, lastTerm = rf.lastLogSignature()
	if args.Term >= rf.CurrentTerm && (rf.VotedFor == unvoted || rf.VotedFor == args.CandidateId) && rf.isUpToDate(args.LastLogIndex, args.LastLogTerm) {
		rf.VotedFor = args.CandidateId
		rf.resetElectionTimeout()
		reply.VoteGranter = true
	}
	rf.LogClass(LOG_CLASS_VOTE, "回复%t 给 %d,给%d投票（last[%d]:%d):%+v", reply.VoteGranter, args.CandidateId, rf.VotedFor, lastIndex, lastTerm, args)
}

// Start handles client requests for the leader node
func (rf *Raft) Start(command LogValue) (int, int, bool) {
	index := 1
	isLeader := rf.State == leader
	term := rf.CurrentTerm
	rf.InLock(func() {
		term = rf.CurrentTerm
		isLeader = rf.State == leader
		if isLeader {
			rf.LogClass(LOG_CLASS_CLIENT, "收到客户端请求:%+v", command)
			index = rf.createLogEntry(command)
			rf.LogClass(LOG_CLASS_CLIENT, "回复请求%+v:%d", command, index)
		}
	}, "Start")
	return index, term, isLeader
}

func (rf *Raft) createLogEntry(command LogValue) int {
	index := -1
	if rf.State == leader {
		rf.LogClass(LOG_CLASS_LIFECYCLE, "创建日志:%+v", command)
		entry := LogEntry{
			Term:  rf.CurrentTerm,
			Value: command,
		}
		rf.Logs = append(rf.Logs, entry)
		lastLogIndex := rf.len()
		rf.NextIndex[rf.Me] = lastLogIndex + 1
		rf.MatchIndex[rf.Me] = lastLogIndex
		index = lastLogIndex
		rf.heartbeat()
		rf.LogClass(LOG_CLASS_CLIENT, "创建日志 %+v:%d", command, index)
	}
	return index
}

// 没有用的调试函数，关闭消息通道
func (rf *Raft) Kill() {
	rf.LogClass(LOG_CLASS_LIFECYCLE, "killed")
	close(rf.KillCh)
}
func (rf *Raft) isConnected() bool {
	connectCount := 0
	for i, v := range rf.Connected {
		if v || i == rf.Me {
			connectCount++
		}
	}
	return connectCount*2 > 3
}
